package com.hgw.easyrocketmq.batch;

import org.apache.rocketmq.common.message.Message;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * Description: MQ消息列表分割迭代器
 * 因MQ批量消息的总大小不应超过4MB，故编写此分割迭代器。
 *
 * @author LinHuiBa-YanAn
 * @date 2023/2/14 13:43
 */
public class MqListSplitter implements Iterator<List<Message>> {

    /**
     * 消息体大小-4kb
     */
    private final Integer MSG_SIZE_LIMIT = 1024 * 4;
    private final List<Message> messages;
    private int currIndex;

    public MqListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    /**
     * Returns true if the iteration has more elements. (In other words, returns true if next would return an element rather than throwing an exception.)
     *
     * @return true if the iteration has more elements
     */
    @Override
    public boolean hasNext() {
        return currIndex < messages.size();
    }

    /**
     * Returns the next element in the iteration.
     *
     * @return the next element in the iteration
     */
    @Override
    public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        while (nextIndex < messages.size()) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            // 增加日志的开销20字节
            tmpSize = tmpSize + 20;
            if (tmpSize > MSG_SIZE_LIMIT) {
                //单个消息超过了最大的限制
                //忽略,否则会阻塞分裂的进程
                if (nextIndex - currIndex == 0) {
                    //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > MSG_SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
            nextIndex++;
        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
}
